package com.heima.article.stream;

import com.heima.article.config.KafkaStreamListener;
import com.heima.common.contants.MQConstants;
import com.heima.model.article.dtos.ArticleVisitStreamMsg;
import com.heima.model.article.dtos.UpdateArticleMsg;
import com.heima.utils.common.JsonUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.stereotype.Component;
import springfox.documentation.spring.web.json.Json;

import java.time.Duration;
import java.util.Arrays;

/**
 * 热点文章实时计算流处理
 */
@Component
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String,String>> {
    @Override
    public String listenerTopic() {
        return MQConstants.HOT_ARTICLE_INPUT_TOPIC;
    }

    @Override
    public String sendTopic() {
        return MQConstants.HOT_ARTICLE_OUTPUT_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        /**
         * 源消息格式：
         *     key       value
         *            {"articleId":1001,"type":"LIKES"}
         *            {"articleId":1001,"type":"LIKES"}
         *            {"articleId":1002,"type":"LIKES"}
         *            {"articleId":1001,"type":"COMMENT"}
         *            ......
         *
         * 目标消息格式：
         *     key       value
         *             {"articleId":1001,"view":Null,"like":300,"collect":null,"comment":null}
         *             {"articleId":1002,"view":50,"like":null,"collect":null,"comment":null}
         *             .....
         */

        /**
         * 第一步：将源消息格式转换为更加方便统计的格式，如： 1001:LIKES
         *    格式：
         *     key       value
         *             ["1001:LIKES]
         *             ["1001:LIKES "]
         *             ["1002:LIKES "]
         *             ["1002:COMMENT"]
         *
         * 第二步：对key-value进行处理，把key的值替换为value
         *     key                      value
         *   ["1001:LIKES]          ["1001:LIKES]
         *   ["1001:LIKES "]          ["1001:LIKES "]
         *   ["1002:LIKES "]          ["1002:LIKES "]
         *   ["1002:COMMENT"]        ["1002:COMMENT"]
         *
         * 第三步：对key进行分组
         * 第四步：对key分组后的结果进行统计数量
         *     key                      value
         *   1001:LIKES                 100
         *   1002:LIKES                 50
         *   1001:COMMENT               60
         */
        KTable<Windowed<Object>, Long> kTable = stream.flatMapValues(new ValueMapper<String, Iterable<?>>() {
            @Override
            public Iterable<?> apply(String value) {  // value： {"articleId":1001,"type":"LIKES"}
                UpdateArticleMsg articleMsg = JsonUtils.toBean(value, UpdateArticleMsg.class);
                String msgStr = articleMsg.getArticleId() + ":" + articleMsg.getType().name();  // 1001:LIKES
                return Arrays.asList(msgStr);
            }
        }).map(new KeyValueMapper<String, Object, KeyValue<?, ?>>() {

            @Override
            public KeyValue<?, ?> apply(String key, Object value) {
                return new KeyValue<>(value, value);
            }
        }).groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .count(Materialized.as("count"));

        //将KTable转换为KStream
        KStream kStream = kTable.toStream().map(new KeyValueMapper<Windowed<Object>, Long, KeyValue<?, ?>>() {
            @Override
            public KeyValue<?, ?> apply(Windowed<Object> windowed, Long value) {
                String key = (String)windowed.key(); // 1001:LIKES
                String[] split = key.split(":");

                ArticleVisitStreamMsg streamMsg = new ArticleVisitStreamMsg();
                streamMsg.setArticleId(Long.valueOf(split[0]));

                switch (UpdateArticleMsg.UpdateArticleType.valueOf(split[1])){
                    case VIEWS:
                        streamMsg.setView(value);
                        break;
                    case COLLECTION:
                        streamMsg.setCollect(value);
                        break;
                    case COMMENT:
                        streamMsg.setComment(value);
                        break;
                    case LIKES:
                        streamMsg.setLike(value);
                        break;

                }

                return new KeyValue<>(key, JsonUtils.toString(streamMsg));
            }
        });
        return kStream;
    }
}
